/**
    Demo program for common neighbor algorithm
    Copyright (C) 2010 wuyu(wuyu.zju _at_ gmail _dot_ com)

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
**/

#include<iostream>
#include<string>
#include<fstream>
#include<vector>
#include<algorithm>
#include<iterator>

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

using namespace std;

#define threadcount 2
#define batch_size 10000

pthread_t threads[threadcount];

vector<uint64_t> index;
vector<uint32_t> ids;
// minimum thresh hold of counts for a common neighbor to appear in results
int thresh_hold = 200;
int bad_user_count = 10000; // 1000+ friends mean it's spam or so, bad users!
uint64_t total_iteration = 0;
bool undirected = true;

vector<uint32_t> baduser;

pthread_mutex_t rangelock = PTHREAD_MUTEX_INITIALIZER, addlock = PTHREAD_MUTEX_INITIALIZER;
int current_range = 0;

void getfriends(vector<uint64_t>& index, vector<uint32_t>& friends, uint32_t id)
{
    if(binary_search(baduser.begin(), baduser.end(), id)) return;
    uint64_t key = id;
    key <<= 32;
    for(vector<uint64_t>::iterator i = lower_bound(index.begin(), index.end(), key); i != index.end(); ++ i)
    {
        uint32_t id1, id2;
        id1 = *i >> 32;
        id2 = *i;
        if(id1 != id)
            break;
        friends.push_back(id2);
    }
}

void getrange(int& start, int& stop)
{
    pthread_mutex_lock(&rangelock);
    start = current_range;
    stop = current_range + batch_size;
    current_range += batch_size;

    int threadid;
    for(threadid = 0; threadid < threadcount; threadid ++)
    {
        if(pthread_equal(threads[threadid], pthread_self()))
            break;
    }

    cout << "allocate " << start << "/" << (int)ids.size() << " for thread #" << threadid << " @ops:" << total_iteration << endl;
    pthread_mutex_unlock(&rangelock);
}

void additeration(uint64_t toadd)
{
    pthread_mutex_lock(&addlock);
    total_iteration += toadd;
    pthread_mutex_unlock(&addlock);
}

void *cn(void *_)
{
    while(true)
    {
        int start, stop;
        getrange(start, stop);
        if(start >= ids.size())
            break;
        if(stop > ids.size())
            stop = ids.size();

        // calculate the CN one by one
        for(int i=start; i < stop; i++)
        {
            vector<uint32_t> friends, fof;
            getfriends(index, friends, ids[i]);
            for(vector<uint32_t>::iterator itr_fid = friends.begin();
                    itr_fid != friends.end();
                    itr_fid++)
            {
                getfriends(index, fof, *itr_fid);
            }
            additeration((int)fof.size());
            sort(fof.begin(), fof.end());
            uint32_t lastid = -1;
            int count = 0;
            fof.push_back(-1);
            for(vector<uint32_t>::iterator itr_fofid = fof.begin();
                    itr_fofid != fof.end();
                    itr_fofid++)
            {
                uint32_t thisid = * itr_fofid;
                if(thisid != lastid)
                {
                    if(lastid > 0 && count > thresh_hold)
                    {
                        // print out result
                        cout << ids[i] << "," << thisid << "," << count << endl;
                    }
                    lastid = thisid;
                    count = 0;
                }
                count ++;
            }
        }
    }
}

int main()
{
    ifstream infile("../sample/ur_44150000_45150000.txt");
    string sid1,sid2;
    int line = 0;

    while(!getline(infile, sid1, ',').eof())
    {
        line ++;

        getline(infile, sid2);

        uint32_t id1, id2;
        id1 = strtol(sid1.c_str(), NULL, 10);
        id2 = strtol(sid2.c_str(), NULL, 10);

        ids.push_back(id1);
        ids.push_back(id2);

        // use unit64_t to contain id1->id2
        uint64_t idid = id1;
        idid <<= 32;
        idid += id2;
        index.push_back(idid);

        // if undirected, push id2->id1
        if(undirected)
        {
            idid = id2;
            idid <<= 32;
            idid += id1;
            index.push_back(idid);
        }
    }
    // unique-lize ids
    sort(ids.begin(), ids.end());
    ids.resize(unique(ids.begin(), ids.end()) - ids.begin());

    cout << "max id is " << *(ids.end()-1) << endl;
    cout << "total lines read:" << line << endl;
    cout << "total id count:" << (int)ids.size() << endl;
    cout << "total relations:" << (int)index.size() << endl;

    // sort to make the index REAL
    sort(index.begin(), index.end());
    // get rid of duplicates
    index.resize(unique(index.begin(), index.end()) - index.begin());

    // filter bad users
    for(int i=0; i<ids.size(); i++)
    {
        vector<uint32_t> friends;
        getfriends(index, friends, ids[i]);
        if(friends.size() > bad_user_count)
        {
            cout << "bad user #" << ids[i] << " has " << friends.size() << " friends" << endl;
            baduser.push_back(ids[i]);
        }
    }
    sort(baduser.begin(), baduser.end());

    for(int i=0; i<threadcount; i++)
    {
        pthread_create( &threads[i], NULL, cn, NULL);
    }

    for(int i=0; i<threadcount; i++)
    {
        pthread_join( threads[i], NULL);
    }
    cout << "total iterations:" << total_iteration << endl;

    return 0;
}

